package org.kafka.demo.first;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Created by lazy on 16-11-2.
 */
public class KafkaProducer extends Thread {
    private static Logger logger = LoggerFactory.getLogger(KafkaProducer.class);
    private String topic;

    public KafkaProducer(String topic){
        super();
        this.topic = topic;
    }

    @Override
    public void run() {
        Producer producer = createProducer();
        int i=0;
        long time = System.currentTimeMillis();
        logger.info("start {}", time);
        while(true){
            producer.send(new KeyedMessage<Integer, String>(topic, "message:" + i++ + "[header:{msgSeq:14,msgId:22,devId:3870010077,bodyLen:6},body:{intervalAccOn:600,intervalAccOff:6300,intervalAccOffMin:105},header:{msgSeq:14,msgId:22,devId:3870010077,bodyLen:6},body:{intervalAccOn:600,intervalAccOff:6300,intervalAccOffMin:105},header:{msgSeq:14,msgId:22,devId:3870010077,bodyLen:6},body:{intervalAccOn:600,intervalAccOff:6300,intervalAccOffMin:105},header:{msgSeq:14,msgId:22,devId:3870010077,bodyLen:6},body:{intervalAccOn:600,intervalAccOff:6300,intervalAccOffMin:105}]"));
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            if (i % 10000 == 0) {
                logger.info("send {}, time {}ms", i, (System.currentTimeMillis() - time));
                time = System.currentTimeMillis();
            }

        }
    }

    private Producer createProducer() {
        Properties properties = new Properties();
//        properties.put("zookeeper.connect", "192.168.1.110:2181,192.168.1.111:2181,192.168.1.112:2181"); //声明zk
        properties.put("zookeeper.connect", "192.168.0.189:2181");
        properties.put("serializer.class", StringEncoder.class.getName());
        // properties.put("metadata.broker.list", "192.168.1.110:9092,192.168.1.111:9093,192.168.1.112:9094");// 声明kafka broker
        properties.put("metadata.broker.list", "192.168.0.189:9092");
        return new Producer<Integer, String>(new ProducerConfig(properties));
    }

    public static void main(String[] args) {
        new KafkaProducer("mytopic").start();// 使用kafka集群中创建好的主题 test
    }
}
